package com.flink.wc.demo.demo_20241225;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

public class FlinkWirteMysql {
    // flink 写入  mysql
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 创建集合数据
        List<String> list = Arrays.asList(
                "192.168.116.141\t1601297294548\tPOST\taddOrder",
                "192.168.116.142\t1601297294549\tGET\tgetOrder"
        );
        env.fromCollection(list).addSink(JdbcSink.sink(
                // 配置SQL语句
                "insert into flink_test(ip, time, type, api) values(?, ?, ?, ?)",
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        System.out.println("receive ====> " + s);
                        // 解析数据
                        String[] elements = s.split("\t");
                        // 新增数据
                        preparedStatement.setString(1, elements[0]);
                        preparedStatement.setString(2, elements[1]);
                        preparedStatement.setString(3, elements[2]);
                        preparedStatement.setString(4, elements[3]);
                    }
                },
                // 配置数据库连接选项
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        // MySQL 数据库 URL
                        .withUrl("jdbc:mysql://localhost:3306/mysql")
                        // MySQL JDBC 驱动
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        // 用户名
                        .withUsername("root")
                        // 密码
                        .withPassword("123456")
                        .build()
        ));

        // 执行 Flink 作业
        env.execute("Flink JDBC Sink to MySQL Example");
    }
}
